package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Demo04DSL {
  def main(args: Array[String]): Unit = {
    /**
     * SQL常见的操作：
     * select
     * from
     * where 过滤
     * group by 分组 ---> 聚合函数 sum、max、min、count、avg
     * having 用于聚合之后的过滤
     * order by 排序
     * limit 限制结果的条数
     * 关联：
     * 内连接：join(inner join)
     * 外连接：outer join
     * 左外连接：left outer join (left join)
     * 右外连接：right outer join (right join)
     * 全外连接：full outer join (full join)
     * 拼接：
     * union : 去重并拼接
     * union all : 不会去重
     * 函数：
     * 条件函数：if、case when、coalesce
     * 字符串函数：split、substring、concat、concat_ws、......
     * 聚合函数：sum、max、min、count、avg
     * 日期函数：日期与字符串之间的相互转换
     * 窗口函数：row_number、rank、dense_rank、lag、lead、......
     * 其他函数
     * */

    // 创建SparkSession
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo04DSL")
      .master("local")
      .config("spark.sql.shuffle.partitions", "2")
      .getOrCreate()

    // 导入Spark SQL的隐式转换 和 SparkSQL提供的方法
    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 读取学生信息数据
    val stuDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id string,name string,age int,gender string,clazz string")
      .load("Spark/data/students.txt")

    // 对多次使用的DF可以进行缓存
    stuDF.cache()

    /**
     * DSL : 领域特定语言 Domain Specific Language
     * 在Spark SQL中实际上就是 基于DataFrame的API
     * 相比RDD而言，DF提供了很多好用的方法 可以让数据分析更加简单方便
     * 实际上 DataFrame 同 DataSet[Row]
     * DF中每一行数据都是一个Row对象（拥有结构）
     */


    // select 取出结果列
    stuDF.select("id", "name", "age")
    //      .show() // 字符串表达式 只有在运行时才能发现错误
    stuDF.select($"id", $"name", $"age")
    //      .show() // 通过 $取列 --> 列表达式（推荐） 可以检测出 语法的错误

    // where 过滤
    // 过滤年龄 > 23 所有的学生
    stuDF.where("age>23")
    //      .show() // 字符串表达式
    stuDF.where($"age" > 23)
    //      .show() // 列表达式

    // 过滤出 文科班的学生
    stuDF.where("clazz like '文科%'")
    //      .show()
    stuDF.where($"clazz" like "文科%")
    //      .show()

    /**
     * 还可以使用filter进行过滤
     * 1、可以传入字符串表达式
     * 2、可以传入列表达式
     * 3、可以传入匿名函数
     */
    // 过滤性别为男的学生
    stuDF.filter("gender='男'").show()
    stuDF.filter($"gender" === "男").show()
    stuDF.filter((row: Row) => {
      // row是Row的对象 表示一条数据 有结构
      val gender: String = row.getAs[String]("gender")
      gender == "男"
    })
    //      .show()


    // groupBy分组 聚合
    /**
     * 分组聚合 会产生shuffle 会导致任务数量过多
     * 可以设置参数：spark.sql.shuffle.partitions 默认值 200
     * 当数据量不大时 可以将该值减小
     */
    // 统计每个班级的性别人数
    stuDF
      .groupBy($"clazz", $"gender")
      .agg(count("id"))
    //      .show()

    stuDF
      .groupBy($"clazz", $"gender")
      .agg(countDistinct("id")) // count统计数量时一定要注意数据是否有重复
    //      .show()

    // having 分组之后的过滤
    // 在DSL中没有having 直接使用where进行过滤即可
    // 统计班级人数 把人数大于80的班级过滤出来
    stuDF
      .groupBy("clazz")
      .agg(countDistinct("id") as "cnt")
      .where($"cnt" > 80)
    //      .show()

    // 排序
    // 统计班级人数 按照人数降序展示前3名
    stuDF
      .groupBy("clazz")
      .agg(countDistinct("id") as "cnt")
      .orderBy($"cnt".desc)
      .limit(3)
    //      .show()

    // 关联
    val scoreDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id String,subject_id String,score Int")
      .load("Spark/data/score.txt")

    // 内连接 join/inner join
    // 当两个DataFrame关联的字段名一样时直接指定关联的字段即可
    stuDF.join(scoreDF, "id")
    //      .show(100)

    scoreDF.printSchema() // 打印DF的结构

    val newScoreDF: DataFrame = scoreDF.withColumnRenamed("id", "new_id")

    newScoreDF.printSchema()

    // 当两个DataFrame关联的字段名不一样时需要使用列表达式指定
    stuDF.join(newScoreDF, $"id" === $"new_id")
    //      .show(50)

    // 外连接：left join、right join、full join

    // 两个DF关联的字段名不同
    // 左外连接：left join/left outer join
    stuDF.join(newScoreDF, $"id" === $"new_id", "left")
    //      .show(20)
    // 右外连接：right join/right outer join
    stuDF.join(newScoreDF, $"id" === $"new_id", "right")
    //      .show(20)
    // 全外连接：full join/full outer join
    stuDF.join(newScoreDF, $"id" === $"new_id", "full")
      .where($"new_id" === "1500100001")
    //      .show(20)

    // 两个DF关联的字段名相同 不能通过列表达式指定关联条件
    // 需要将关联的字段 构建成List作为参数传入，才能够指定关联的类型
    //    stuDF.join(scoreDF, $"id" === $"id").show() // 会报错：Reference 'id' is ambiguous, could be: id, id.;
    stuDF.join(scoreDF, List("id"), "left")
    //      .show(15)
    stuDF.join(scoreDF, List("id"), "right")
    //      .show(15)
    stuDF.join(scoreDF, List("id"), "full")
    //      .show(15)

    // 拼接 只能对相同结构的两个DF进行拼接
    // 对stu数据进行抽样
    val stuSampleDF1: DataFrame = stuDF.sample(withReplacement = false, fraction = 0.01, seed = 1)
    val stuSampleDF2: Dataset[Row] = stuDF.sample(withReplacement = false, fraction = 0.01, seed = 1)
    stuSampleDF1.union(stuSampleDF2)
    //      .show(50) // 相当于SQL中union all 不进行去重
    stuSampleDF1.union(stuSampleDF2).distinct()
    //      .show(50) // 相当于SQL中union 会去重

    // 条件函数：if、case when、coalesce
    // 给每个同学的成绩进行一个评级
    /**
     * 大于等于90 ：优秀
     * 大于等于80 ：良好
     * 大于等于70 ：一般
     * 大于等于60 ：及格
     * 小于60 ：不及格
     */
    // 加载科目数据，将学生成绩转化成百分制
    val subjectDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("subject_id String,subject_name String,subject_score Int")
      .load("Spark/data/subject.txt")

    scoreDF
      .join(subjectDF, List("subject_id"), "left")
      .select($"id", $"subject_id", round($"score" * 100.toDouble / $"subject_score", 2) as "score")
      // 如果需要额外增加一列可以使用withColumn
      .withColumn("cls",
        when($"score" >= 90, "优秀")
          .when($"score" >= 80, "良好")
          .when($"score" >= 70, "一般")
          .when($"score" >= 60, "及格")
          .otherwise("不及格")
      )
      .show()

    //    stuDF.select(coalesce($"id", $"name", $"age", $"gender"))

    // 字符串函数
    stuDF
      .withColumn("col", split($"clazz", "科")(1))
      .withColumn("col2", md5($"id"))
      .show()

    // 日期函数 日期与字符串之间的转换

    // 加载电信数据
    val dianXinDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("mdn String,grid_id Bigint,city_id Int,county_id Int,duration Int,start_time String,end_time String,pt String")
      .load("Spark/data/dianxin.csv")

    dianXinDF.printSchema()
    dianXinDF.show()

    // 将start_time 转换成 datetime 类型
    dianXinDF
      .select($"mdn", $"start_time", from_unixtime(unix_timestamp($"start_time", "yyyyMMddHHmmss"), "yyyy年MM/dd HH-mm:ss"))
      .show()


    // 基于电信数据找出 每个城市游客数最多的前三个区县 --> 组内TopN

    dianXinDF.createOrReplaceTempView("dianxin")

    // SQL的方式
    spark
      .sql(
        """
          |SELECT  tt1.city_id
          |        ,tt1.county_id
          |        ,tt1.cnt
          |        ,tt1.rn
          |from (
          |    select  t1.city_id
          |            ,t1.county_id
          |            ,t1.cnt
          |            ,row_number() over (partition by t1.city_id order by t1.cnt desc) as rn
          |    from (
          |        select  city_id
          |                ,county_id
          |                ,count(distinct mdn) as cnt
          |        from dianxin
          |        group by city_id,county_id
          |    ) t1
          |) tt1 where tt1.rn <= 3
          |""".stripMargin).show()

    // DSL的方式
    dianXinDF
      .groupBy($"city_id", $"county_id")
      .agg(countDistinct("mdn") as "cnt")
      .withColumn("rn", row_number() over Window.partitionBy("city_id").orderBy($"cnt".desc))
      .where($"rn" <= 3)
      .show()

    // 用完记得释放缓存
    stuDF.unpersist()


  }

}
